package com.jxb.namesrv;

import com.jxb.ThreadFactoryImpl;
import com.jxb.exception.RemotingCommandException;
import com.jxb.netty.*;
import io.netty.channel.ChannelHandlerContext;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class NamesrvStartup {

    private static final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable = new HashMap<>();

    public static void main(String[] args) throws InterruptedException {
        DefaultRequestProcessor defaultRequestProcessor = new DefaultRequestProcessor();
        //监听不活跃的broker
        Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
                "NSScheduledThread")).scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                scanNotActiveBroker();
            }
        },5,10, TimeUnit.SECONDS);
        //启动namesrv服务端
        NettyRemotingServer server = new NettyRemotingServer(9875,"namesrv server started success !");
        ExecutorService executorService = Executors.newFixedThreadPool(8, new ThreadFactoryImpl("RemotingExecutorThread_"));
        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(defaultRequestProcessor,executorService);
        server.defaultRequestProcessor = pair;
        server.start();
    }

    private static void scanNotActiveBroker() {
        System.out.println("enter scan not active broker............");
        Iterator<Map.Entry<String, BrokerLiveInfo>> it = brokerLiveTable.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, BrokerLiveInfo> next = it.next();
            long last = next.getValue().getLastUpdateTimestamp();
            if ((last + 1000 * 60 * 2) < System.currentTimeMillis()) {
                //关闭通道
                RemotingUtil.closeChannel(next.getValue().getChannel());
                //移除缓存
                it.remove();
                brokerLiveTable.remove(next.getKey());
            }
        }
    }

    public static class DefaultRequestProcessor implements NettyRequestProcessor {

        @Override
        public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
            switch (request.getCode()) {
                case RequestCode.REGISTER_BROKER:
                    return this.registerBroker(ctx, request);
            }
            return null;
        }

        private RemotingCommand registerBroker(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
            final RegisterBrokerRequestHeader requestHeader =
                    (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);
            DataVersion dataVersion = new DataVersion();
            AtomicLong counter = new AtomicLong(1);
            dataVersion.setCounter(counter);
            dataVersion.setTimestamp(1);
            BrokerLiveInfo prevBrokerLiveInfo = brokerLiveTable.put(requestHeader.getBrokerAddr(),
                    new BrokerLiveInfo(
                            System.currentTimeMillis(),
                            dataVersion,
                            ctx.channel(),
                            requestHeader.getHaServerAddr()));

            RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
            response.setBody("123".getBytes(StandardCharsets.UTF_8));
            response.setCode(200);
            response.setRemark(null);
            return response;
        }

        @Override
        public boolean rejectRequest() {
            return false;
        }

    }
}
